package com.lbjhan.rpc.consumer.client;

import com.lbjhan.rpc.common.RpcRequest;
import com.lbjhan.rpc.common.RpcResponse;
import com.lbjhan.rpc.consumer.handler.RpcNettyClientHandler;
import com.lbjhan.rpc.consumer.handler.RpcNettyClientResponseHandler;
import com.lbjhan.rpc.consumer.handler.RpcNettyClientSerializerDecoder;
import com.lbjhan.rpc.consumer.handler.RpcNettyClientSerializerEncoder;
import com.lbjhan.rpc.handler.RpcCommonDeCoder;
import com.lbjhan.rpc.handler.RpcCommonEncoder;
import com.lbjhan.rpc.serializer.FastjsonSerializer;
import com.lbjhan.rpc.serializer.JacksonSerializer;
import com.lbjhan.rpc.serializer.KryoSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 基于Netty的客户端启动类
 * 单例模式   长链接
 * 1. 启动客户端
 * 2. 发送消息
 * 3. 关闭客户端
 */
public class RpcNettyResponseClient {

    private String ip;
    private int port;
    private NioEventLoopGroup group;
    private Channel channel;
    private RpcNettyClientResponseHandler rpcNettyClientResponseHandler = new RpcNettyClientResponseHandler();
    private ExecutorService executorService = Executors.newCachedThreadPool();

    private static volatile RpcNettyResponseClient rpcNettyResponseClient = null;

    private RpcNettyResponseClient(String ip_port){
        String[] split = ip_port.split(":");
        this.ip = split[0];
        this.port = Integer.parseInt(split[1]);
        initClient();
    }

    public static RpcNettyResponseClient getInstance(String ip_port){
        if(rpcNettyResponseClient == null){
            synchronized (RpcNettyResponseClient.class){
                if(rpcNettyResponseClient == null){
                    rpcNettyResponseClient = new RpcNettyResponseClient(ip_port);
                }
            }
        }
        return rpcNettyResponseClient;
    }

    /**
     * 1. 启动客户端
     */
    private void initClient(){
        try{
            //1. 创建线程组
            group = new NioEventLoopGroup();
            //2. 创建启动助手
            Bootstrap bootstrap = new Bootstrap();
            //3. 设置参数
            bootstrap.group(group).channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //String 类型编码
                            //序列化
//                            pipeline.addLast(new RpcNettyClientSerializerEncoder(new JacksonSerializer<RpcRequest>()));
//                            pipeline.addLast(new RpcNettyClientSerializerDecoder(new JacksonSerializer<RpcResponse>()));
                            pipeline.addLast(new RpcCommonEncoder(new KryoSerializer(), 0, 5));
                            pipeline.addLast(new RpcCommonDeCoder());
                            //业务处理
                            pipeline.addLast(rpcNettyClientResponseHandler);
                        }
                    });
            //4. 连接Netty服务器端
            channel = bootstrap.connect(ip, port).sync().channel();
            System.out.println("客户端连接成功");
        }catch (Exception e){
            e.printStackTrace();
            if(null != channel)
                channel.close();
            if(null != group)
                group.shutdownGracefully();
        }

    }

    /**
     * 2. 发送消息到服务器端
     * @param rpcRequest
     * @return
     */
    public Object send(RpcRequest rpcRequest) throws ExecutionException, InterruptedException {
        rpcNettyClientResponseHandler.setRequestMsg(rpcRequest);
        Future submit = executorService.submit(rpcNettyClientResponseHandler);
        return submit.get();
    }

    /**
     * 关闭客户端
     */
    public void close(){
        if(null != channel)
            channel.close();
        if(null != group)
            group.shutdownGracefully();
    }

}
